"""
多用户并发流管理服务
"""

import asyncio
import time
from typing import Dict, List, Any, Optional, Set, Tuple
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict, deque
import threading
import weakref

from ...core.exceptions import BusinessLogicError, ValidationError
from ...core.logging import get_logger
from ...core.config import settings
from ...services.cache import cache_service

logger = get_logger(__name__)


class ConcurrencyLevel(Enum):
    """并发级别"""
    LOW = "low"           # 低并发：每用户1-2个流
    MEDIUM = "medium"     # 中并发：每用户3-5个流
    HIGH = "high"         # 高并发：每用户6-10个流
    UNLIMITED = "unlimited" # 无限制


class Priority(Enum):
    """优先级"""
    LOW = 1
    NORMAL = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class StreamResource:
    """流资源"""
    stream_id: str
    user_id: str
    session_id: str
    priority: Priority = Priority.NORMAL
    bandwidth_requirement: float = 128.0  # kbps
    cpu_requirement: float = 1.0          # CPU核心数
    memory_requirement: float = 50.0      # MB
    created_at: float = field(default_factory=time.time)
    last_accessed: float = field(default_factory=time.time)


@dataclass
class ResourceLimits:
    """资源限制"""
    max_concurrent_streams: int = 100
    max_streams_per_user: Dict[str, int] = field(default_factory=lambda: {
        ConcurrencyLevel.LOW.value: 2,
        ConcurrencyLevel.MEDIUM.value: 5,
        ConcurrencyLevel.HIGH.value: 10,
        ConcurrencyLevel.UNLIMITED.value: 999
    })
    total_bandwidth_limit: float = 10000.0  # kbps
    total_cpu_limit: float = 8.0            # CPU核心数
    total_memory_limit: float = 4096.0       # MB
    user_bandwidth_limit: float = 1000.0     # kbps per user
    user_cpu_limit: float = 2.0              # CPU核心数 per user
    user_memory_limit: float = 512.0         # MB per user


@dataclass
class ResourceUsage:
    """资源使用情况"""
    active_streams: int = 0
    total_bandwidth: float = 0.0
    total_cpu: float = 0.0
    total_memory: float = 0.0
    user_stream_counts: Dict[str, int] = field(default_factory=dict)
    user_bandwidth_usage: Dict[str, float] = field(default_factory=dict)
    user_cpu_usage: Dict[str, float] = field(default_factory=dict)
    user_memory_usage: Dict[str, float] = field(default_factory=dict)


class ConcurrencyManager:
    """并发管理器"""
    
    def __init__(self, limits: Optional[ResourceLimits] = None):
        self.limits = limits or ResourceLimits()
        
        # 资源跟踪
        self.active_streams: Dict[str, StreamResource] = {}
        self.user_streams: Dict[str, Set[str]] = defaultdict(set)  # user_id -> {stream_ids}
        self.usage = ResourceUsage()
        
        # 优先级队列
        self.pending_streams: asyncio.Queue = asyncio.Queue()
        self.admission_queue: asyncio.Queue = asyncio.Queue()
        
        # 控制标志
        self._admission_running = False
        self._monitoring_running = False
        self._shutdown = False
        
        # 回调函数
        self._stream_admitted_callbacks: List[callable] = []
        self._stream_rejected_callbacks: List[callable] = []
        self._resource_warning_callbacks: List[callable] = []
        
        # 统计信息
        self.stats = {
            "total_requests": 0,
            "admitted_requests": 0,
            "rejected_requests": 0,
            "preempted_requests": 0,
            "average_wait_time": 0.0,
            "peak_concurrent_streams": 0
        }
        
        logger.info("ConcurrencyManager initialized")
    
    async def start(self):
        """启动并发管理器"""
        if self._admission_running:
            return
        
        self._admission_running = True
        self._monitoring_running = True
        self._shutdown = False
        
        # 启动准入控制任务
        asyncio.create_task(self._admission_controller())
        
        # 启动监控任务
        asyncio.create_task(self._resource_monitor())
        
        logger.info("ConcurrencyManager started")
    
    async def stop(self):
        """停止并发管理器"""
        self._shutdown = True
        self._admission_running = False
        self._monitoring_running = False
        
        logger.info("ConcurrencyManager stopped")
    
    async def request_stream(
        self,
        stream_id: str,
        user_id: str,
        session_id: str,
        priority: Priority = Priority.NORMAL,
        bandwidth_requirement: float = 128.0,
        cpu_requirement: float = 1.0,
        memory_requirement: float = 50.0,
        concurrency_level: ConcurrencyLevel = ConcurrencyLevel.MEDIUM
    ) -> Tuple[bool, str]:
        """请求流资源"""
        try:
            # 创建流资源
            resource = StreamResource(
                stream_id=stream_id,
                user_id=user_id,
                session_id=session_id,
                priority=priority,
                bandwidth_requirement=bandwidth_requirement,
                cpu_requirement=cpu_requirement,
                memory_requirement=memory_requirement
            )
            
            # 检查用户并发限制
            if not await self._check_user_limits(user_id, concurrency_level):
                msg = f"User {user_id} exceeded concurrency limit"
                await self._notify_stream_rejected(resource, msg)
                return False, msg
            
            # 添加到准入队列
            await self.admission_queue.put((resource, concurrency_level))
            self.stats["total_requests"] += 1
            
            logger.info(f"Stream request submitted: {stream_id} for user {user_id}")
            return True, "Request submitted"
            
        except Exception as e:
            logger.error(f"Failed to request stream: {e}")
            return False, str(e)
    
    async def release_stream(self, stream_id: str) -> bool:
        """释放流资源"""
        try:
            if stream_id not in self.active_streams:
                logger.warning(f"Stream {stream_id} not found in active streams")
                return False
            
            resource = self.active_streams[stream_id]
            user_id = resource.user_id
            
            # 从活动流中移除
            del self.active_streams[stream_id]
            
            # 从用户流列表中移除
            if user_id in self.user_streams:
                self.user_streams[user_id].discard(stream_id)
                if not self.user_streams[user_id]:
                    del self.user_streams[user_id]
            
            # 更新使用统计
            self._update_usage_on_release(resource)
            
            logger.info(f"Stream {stream_id} released for user {user_id}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to release stream: {e}")
            return False
    
    async def update_stream_priority(self, stream_id: str, new_priority: Priority) -> bool:
        """更新流优先级"""
        try:
            if stream_id not in self.active_streams:
                return False
            
            old_priority = self.active_streams[stream_id].priority
            self.active_streams[stream_id].priority = new_priority
            
            logger.info(f"Stream {stream_id} priority updated: {old_priority.value} -> {new_priority.value}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to update stream priority: {e}")
            return False
    
    async def get_user_streams(self, user_id: str) -> List[StreamResource]:
        """获取用户的流"""
        if user_id not in self.user_streams:
            return []
        
        stream_ids = self.user_streams[user_id]
        return [
            self.active_streams[stream_id] 
            for stream_id in stream_ids 
            if stream_id in self.active_streams
        ]
    
    async def get_resource_usage(self) -> ResourceUsage:
        """获取资源使用情况"""
        self._update_usage()
        return self.usage
    
    async def get_statistics(self) -> Dict[str, Any]:
        """获取统计信息"""
        current_usage = await self.get_resource_usage()
        
        stats = {
            **self.stats,
            "current_active_streams": current_usage.active_streams,
            "current_bandwidth_usage": current_usage.total_bandwidth,
            "current_cpu_usage": current_usage.total_cpu,
            "current_memory_usage": current_usage.total_memory,
            "active_users": len(self.user_streams),
            "resource_utilization": {
                "bandwidth": (current_usage.total_bandwidth / self.limits.total_bandwidth_limit) * 100,
                "cpu": (current_usage.total_cpu / self.limits.total_cpu_limit) * 100,
                "memory": (current_usage.total_memory / self.limits.total_memory_limit) * 100,
                "streams": (current_usage.active_streams / self.limits.max_concurrent_streams) * 100
            }
        }
        
        return stats
    
    async def enforce_user_limits(self, user_id: str, max_streams: int) -> int:
        """强制执行用户限制"""
        user_streams = await self.get_user_streams(user_id)
        
        if len(user_streams) <= max_streams:
            return 0
        
        # 按优先级排序，低优先级的流先被移除
        sorted_streams = sorted(user_streams, key=lambda x: x.priority.value)
        
        # 移除超出限制的流
        streams_to_remove = len(user_streams) - max_streams
        removed_count = 0
        
        for stream in sorted_streams[:streams_to_remove]:
            if await self.release_stream(stream.stream_id):
                removed_count += 1
                logger.info(f"Preempted stream {stream.stream_id} for user {user_id}")
                self.stats["preempted_requests"] += 1
        
        return removed_count
    
    def add_stream_admitted_callback(self, callback: callable):
        """添加流准入回调"""
        self._stream_admitted_callbacks.append(callback)
    
    def add_stream_rejected_callback(self, callback: callable):
        """添加流拒绝回调"""
        self._stream_rejected_callbacks.append(callback)
    
    def add_resource_warning_callback(self, callback: callable):
        """添加资源警告回调"""
        self._resource_warning_callbacks.append(callback)
    
    async def _check_user_limits(
        self, 
        user_id: str, 
        concurrency_level: ConcurrencyLevel
    ) -> bool:
        """检查用户限制"""
        current_count = len(self.user_streams.get(user_id, set()))
        max_allowed = self.limits.max_streams_per_user.get(concurrency_level.value, 5)
        
        return current_count < max_allowed
    
    async def _check_resource_availability(self, resource: StreamResource) -> bool:
        """检查资源可用性"""
        # 检查总并发流限制
        if self.usage.active_streams >= self.limits.max_concurrent_streams:
            return False
        
        # 检查带宽限制
        if (self.usage.total_bandwidth + resource.bandwidth_requirement > 
            self.limits.total_bandwidth_limit):
            return False
        
        # 检查CPU限制
        if (self.usage.total_cpu + resource.cpu_requirement > 
            self.limits.total_cpu_limit):
            return False
        
        # 检查内存限制
        if (self.usage.total_memory + resource.memory_requirement > 
            self.limits.total_memory_limit):
            return False
        
        return True
    
    async def _check_user_resource_limits(self, resource: StreamResource) -> bool:
        """检查用户资源限制"""
        user_id = resource.user_id
        
        # 检查用户带宽限制
        user_bandwidth = self.usage.user_bandwidth_usage.get(user_id, 0)
        if (user_bandwidth + resource.bandwidth_requirement > 
            self.limits.user_bandwidth_limit):
            return False
        
        # 检查用户CPU限制
        user_cpu = self.usage.user_cpu_usage.get(user_id, 0)
        if (user_cpu + resource.cpu_requirement > 
            self.limits.user_cpu_limit):
            return False
        
        # 检查用户内存限制
        user_memory = self.usage.user_memory_usage.get(user_id, 0)
        if (user_memory + resource.memory_requirement > 
            self.limits.user_memory_limit):
            return False
        
        return True
    
    async def _admission_controller(self):
        """准入控制器"""
        while not self._shutdown and self._admission_running:
            try:
                # 等待准入请求
                resource, concurrency_level = await asyncio.wait_for(
                    self.admission_queue.get(), timeout=1.0
                )
                
                # 检查准入条件
                can_admit = await self._check_resource_availability(resource)
                can_admit = can_admit and await self._check_user_resource_limits(resource)
                can_admit = can_admit and await self._check_user_limits(
                    resource.user_id, concurrency_level
                )
                
                if can_admit:
                    # 准入流
                    await self._admit_stream(resource)
                    self.stats["admitted_requests"] += 1
                else:
                    # 检查是否可以抢占低优先级流
                    if await self._try_preemption(resource):
                        await self._admit_stream(resource)
                        self.stats["admitted_requests"] += 1
                    else:
                        # 拒绝流
                        await self._reject_stream(resource, "Resource limits exceeded")
                        self.stats["rejected_requests"] += 1
                
            except asyncio.TimeoutError:
                continue
            except Exception as e:
                logger.error(f"Admission controller error: {e}")
                await asyncio.sleep(1.0)
    
    async def _try_preemption(self, new_resource: StreamResource) -> bool:
        """尝试抢占资源"""
        # 获取比新资源优先级低的流
        lower_priority_streams = [
            stream for stream in self.active_streams.values()
            if stream.priority.value < new_resource.priority.value
        ]
        
        if not lower_priority_streams:
            return False
        
        # 按优先级和创建时间排序（先抢占低优先级，再抢占早创建的）
        lower_priority_streams.sort(key=lambda x: (x.priority.value, x.created_at))
        
        # 尝试释放足够的资源
        required_bandwidth = new_resource.bandwidth_requirement
        required_cpu = new_resource.cpu_requirement
        required_memory = new_resource.memory_requirement
        
        released_streams = []
        freed_bandwidth = 0.0
        freed_cpu = 0.0
        freed_memory = 0.0
        
        for stream in lower_priority_streams:
            # 检查是否是同一用户（不允许抢占同用户的流）
            if stream.user_id == new_resource.user_id:
                continue
            
            # 释放流
            if await self.release_stream(stream.stream_id):
                released_streams.append(stream)
                freed_bandwidth += stream.bandwidth_requirement
                freed_cpu += stream.cpu_requirement
                freed_memory += stream.memory_requirement
                
                # 检查是否足够
                if (freed_bandwidth >= required_bandwidth and 
                    freed_cpu >= required_cpu and 
                    freed_memory >= required_memory):
                    break
        
        # 如果释放的资源足够，返回成功
        if (freed_bandwidth >= required_bandwidth and 
            freed_cpu >= required_cpu and 
            freed_memory >= required_memory):
            
            for stream in released_streams:
                logger.info(f"Preempted stream {stream.stream_id} for user {stream.user_id}")
                self.stats["preempted_requests"] += 1
            
            return True
        
        # 如果不够，恢复释放的流（这里简化处理，实际中可能需要更复杂的逻辑）
        for stream in released_streams:
            # 重新添加到活动流
            self.active_streams[stream.stream_id] = stream
            self.user_streams[stream.user_id].add(stream.stream_id)
        
        return False
    
    async def _admit_stream(self, resource: StreamResource):
        """准入流"""
        # 添加到活动流
        self.active_streams[resource.stream_id] = resource
        self.user_streams[resource.user_id].add(resource.stream_id)
        
        # 更新使用统计
        self._update_usage_on_admit(resource)
        
        # 更新峰值统计
        self.stats["peak_concurrent_streams"] = max(
            self.stats["peak_concurrent_streams"],
            len(self.active_streams)
        )
        
        # 通知回调
        await self._notify_stream_admitted(resource)
        
        logger.info(f"Stream admitted: {resource.stream_id}")
    
    async def _reject_stream(self, resource: StreamResource, reason: str):
        """拒绝流"""
        await self._notify_stream_rejected(resource, reason)
        logger.warning(f"Stream rejected: {resource.stream_id} - {reason}")
    
    async def _resource_monitor(self):
        """资源监控"""
        while not self._shutdown and self._monitoring_running:
            try:
                # 更新使用统计
                self._update_usage()
                
                # 检查资源使用率
                await self._check_resource_utilization()
                
                # 清理过期流
                await self._cleanup_expired_streams()
                
                # 等待下一次检查
                await asyncio.sleep(10.0)
                
            except Exception as e:
                logger.error(f"Resource monitor error: {e}")
                await asyncio.sleep(5.0)
    
    def _update_usage_on_admit(self, resource: StreamResource):
        """准入时更新使用统计"""
        self.usage.active_streams += 1
        self.usage.total_bandwidth += resource.bandwidth_requirement
        self.usage.total_cpu += resource.cpu_requirement
        self.usage.total_memory += resource.memory_requirement
        
        user_id = resource.user_id
        self.usage.user_stream_counts[user_id] = self.usage.user_stream_counts.get(user_id, 0) + 1
        self.usage.user_bandwidth_usage[user_id] = self.usage.user_bandwidth_usage.get(user_id, 0) + resource.bandwidth_requirement
        self.usage.user_cpu_usage[user_id] = self.usage.user_cpu_usage.get(user_id, 0) + resource.cpu_requirement
        self.usage.user_memory_usage[user_id] = self.usage.user_memory_usage.get(user_id, 0) + resource.memory_requirement
    
    def _update_usage_on_release(self, resource: StreamResource):
        """释放时更新使用统计"""
        self.usage.active_streams -= 1
        self.usage.total_bandwidth -= resource.bandwidth_requirement
        self.usage.total_cpu -= resource.cpu_requirement
        self.usage.total_memory -= resource.memory_requirement
        
        user_id = resource.user_id
        self.usage.user_stream_counts[user_id] = self.usage.user_stream_counts.get(user_id, 0) - 1
        self.usage.user_bandwidth_usage[user_id] = self.usage.user_bandwidth_usage.get(user_id, 0) - resource.bandwidth_requirement
        self.usage.user_cpu_usage[user_id] = self.usage.user_cpu_usage.get(user_id, 0) - resource.cpu_requirement
        self.usage.user_memory_usage[user_id] = self.usage.user_memory_usage.get(user_id, 0) - resource.memory_requirement
        
        # 清理零值
        if self.usage.user_stream_counts[user_id] <= 0:
            del self.usage.user_stream_counts[user_id]
        if self.usage.user_bandwidth_usage.get(user_id, 0) <= 0:
            del self.usage.user_bandwidth_usage[user_id]
        if self.usage.user_cpu_usage.get(user_id, 0) <= 0:
            del self.usage.user_cpu_usage[user_id]
        if self.usage.user_memory_usage.get(user_id, 0) <= 0:
            del self.usage.user_memory_usage[user_id]
    
    def _update_usage(self):
        """更新使用统计"""
        # 重新计算以确保准确性
        self.usage.active_streams = len(self.active_streams)
        self.usage.total_bandwidth = sum(s.bandwidth_requirement for s in self.active_streams.values())
        self.usage.total_cpu = sum(s.cpu_requirement for s in self.active_streams.values())
        self.usage.total_memory = sum(s.memory_requirement for s in self.active_streams.values())
        
        # 重新计算用户使用统计
        self.usage.user_stream_counts.clear()
        self.usage.user_bandwidth_usage.clear()
        self.usage.user_cpu_usage.clear()
        self.usage.user_memory_usage.clear()
        
        for stream in self.active_streams.values():
            user_id = stream.user_id
            
            self.usage.user_stream_counts[user_id] = self.usage.user_stream_counts.get(user_id, 0) + 1
            self.usage.user_bandwidth_usage[user_id] = self.usage.user_bandwidth_usage.get(user_id, 0) + stream.bandwidth_requirement
            self.usage.user_cpu_usage[user_id] = self.usage.user_cpu_usage.get(user_id, 0) + stream.cpu_requirement
            self.usage.user_memory_usage[user_id] = self.usage.user_memory_usage.get(user_id, 0) + stream.memory_requirement
    
    async def _check_resource_utilization(self):
        """检查资源使用率"""
        utilization = {
            "bandwidth": (self.usage.total_bandwidth / self.limits.total_bandwidth_limit) * 100,
            "cpu": (self.usage.total_cpu / self.limits.total_cpu_limit) * 100,
            "memory": (self.usage.total_memory / self.limits.total_memory_limit) * 100,
            "streams": (self.usage.active_streams / self.limits.max_concurrent_streams) * 100
        }
        
        # 检查警告阈值
        warning_threshold = 80.0
        
        for resource_type, usage_percent in utilization.items():
            if usage_percent > warning_threshold:
                await self._notify_resource_warning(resource_type, usage_percent)
    
    async def _cleanup_expired_streams(self):
        """清理过期流"""
        current_time = time.time()
        expired_streams = []
        
        for stream_id, stream in self.active_streams.items():
            # 如果流超过30分钟无访问，标记为过期
            if current_time - stream.last_accessed > 1800:
                expired_streams.append(stream_id)
        
        for stream_id in expired_streams:
            stream = self.active_streams[stream_id]
            await self.release_stream(stream_id)
            logger.info(f"Cleaned up expired stream: {stream_id}")
    
    async def _notify_stream_admitted(self, resource: StreamResource):
        """通知流准入"""
        for callback in self._stream_admitted_callbacks:
            try:
                if asyncio.iscoroutinefunction(callback):
                    await callback(resource)
                else:
                    callback(resource)
            except Exception as e:
                logger.error(f"Stream admitted callback error: {e}")
    
    async def _notify_stream_rejected(self, resource: StreamResource, reason: str):
        """通知流拒绝"""
        for callback in self._stream_rejected_callbacks:
            try:
                if asyncio.iscoroutinefunction(callback):
                    await callback(resource, reason)
                else:
                    callback(resource, reason)
            except Exception as e:
                logger.error(f"Stream rejected callback error: {e}")
    
    async def _notify_resource_warning(self, resource_type: str, usage_percent: float):
        """通知资源警告"""
        for callback in self._resource_warning_callbacks:
            try:
                if asyncio.iscoroutinefunction(callback):
                    await callback(resource_type, usage_percent)
                else:
                    callback(resource_type, usage_percent)
            except Exception as e:
                logger.error(f"Resource warning callback error: {e}")


# 全局并发管理器实例
concurrency_manager = ConcurrencyManager()
